package main

import (
	"context"
	"fmt"
	"net"

	"task/config"
	"task/driver"
	pb "task/pb/business"
	service "task/service/business"

	"github.com/apache/pulsar-client-go/pulsar"
	log "github.com/sirupsen/logrus"
	"google.golang.org/grpc"
)

const channelLen = 100000

func main() {
	var (
		business1Consumer, business2Consumer pulsar.Consumer
		err                                  error
	)
	// 初始化消息队列
	driver.InitPulsarClient()
	defer driver.PulsarClient.Close()
	// 初始化回调消息生产者
	driver.InitCallbackProducer()
	defer driver.CallbackProducer.Close()
	// 创建业务1消息消费者
	business1Channel := make(chan pulsar.ConsumerMessage, channelLen)
	business1Options := pulsar.ConsumerOptions{
		Topic:            config.Conf.Pulsar.Business1Topic,
		SubscriptionName: config.Conf.Pulsar.SubscriptionName,
		Type:             pulsar.Shared,
	}
	business1Options.MessageChannel = business1Channel
	if business1Consumer, err = driver.PulsarClient.Subscribe(business1Options); err != nil {
		log.Panic("Failed to create pulsar business1 consumer:", err)
	}
	defer business1Consumer.Close()
	// 创建业务2消息消费者
	business2Channel := make(chan pulsar.ConsumerMessage, channelLen)
	business2Options := pulsar.ConsumerOptions{
		Topic:            config.Conf.Pulsar.Business2Topic,
		SubscriptionName: config.Conf.Pulsar.SubscriptionName,
		Type:             pulsar.Shared,
	}
	business2Options.MessageChannel = business2Channel
	if business2Consumer, err = driver.PulsarClient.Subscribe(business2Options); err != nil {
		log.Panic("Failed to create pulsar business2 consumer:", err)
	}
	defer business2Consumer.Close()
	// 监听业务1事件消息
	go func() {
		log.Info("Start to monitor business1 message...")
		for cm := range business1Channel {
			msg := cm.Message
			log.Infof("Received business1 message: %s business1 channel len: %d", string(msg.Payload()), len(business1Channel))
			_ = service.StartBusiness1(context.Background(), msg.Payload())
			// 无论成功失败 都要消息确认
			business1Consumer.Ack(msg)
		}
	}()
	// 监听业务2事件消息
	go func() {
		log.Info("Start to monitor business2 message...")
		for cm := range business2Channel {
			msg := cm.Message
			log.Infof("Received business2 message: %s business2 channel len: %d", string(msg.Payload()), len(business2Channel))
			_ = service.StartBusiness2(context.Background(), msg.Payload())
			// 无论成功失败 都要消息确认
			business2Consumer.Ack(msg)
		}
	}()
	// 创建并运行grpc服务
	addr := fmt.Sprintf(":%d", config.Conf.Business.GrpcPort)
	lis, err := net.Listen("tcp", addr)
	if err != nil {
		log.Panic("Failed to listen:", err)
	}
	defer lis.Close()
	grpcServer := grpc.NewServer()
	pb.RegisterBusinessServiceServer(grpcServer, service.NewBusinessServer())
	if err = grpcServer.Serve(lis); err != nil {
		log.Panic("Failed to start business grpc server:", err)
	}
}
